package icasue.runninginvoke.task.adapter;

import com.alibaba.fastjson.JSON;
import icasue.base.common.Const;
import icasue.base.utils.check.CommonCheck;
import icasue.base.utils.wareable.FieldConvertUtil;
import icasue.base.utils.wareable.SnowflakeIdUtil;
import icasue.reflect.handles.fields.FieldOF;
import icasue.reflect.handles.object.ObjectOF;
import icasue.reflect.handles.predicate.SureOF;
import icasue.runninginvoke.config.properties.TasksConfigProperties;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.logging.log4j.util.Strings;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.context.EmbeddedValueResolverAware;
import org.springframework.context.EnvironmentAware;
import org.springframework.core.env.Environment;
import org.springframework.util.StringValueResolver;
import org.springframework.web.client.RestTemplate;
import icasue.runninginvoke.task.inteceptors.RibbonChooseResultInterceptor;
import icasue.runninginvoke.task.proxy.OnceTaskProxy;
import icasue.runninginvoke.task.proxy.ScheduleTaskProxy;
import icasue.runninginvoke.task.proxy.TaskProxyHolder;
import icasue.runninginvoke.task.reader.FileReadUtil;
import icasue.runninginvoke.task.redis.AbstractRedisAsyncTask;
import icasue.runninginvoke.task.redis.AsyncTaskOperate;

import java.lang.invoke.MethodHandle;
import java.lang.invoke.MethodHandles;
import java.lang.invoke.MethodType;
import java.net.InetAddress;
import java.util.*;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.stream.Collectors;

/**
 * @Author: Qiao Hang
 * @CreateDate: 2020/11/10 下午1:26
 * @UpdateDate:
 * @Description:
 */
@Data
@Slf4j
public class TaskAdapterService implements BeanFactoryAware,
            InitializingBean, EmbeddedValueResolverAware, SmartInitializingSingleton, EnvironmentAware {

    private RestTemplate restTemplate;
    private AsyncTaskOperate asyncTaskOperate;
    private BeanFactory beanFactory;
    private StringValueResolver valueResolver;
    private Environment environment;
    private String serviceId;
    private String port;
    private String hostAddress;
    private Boolean balanceServerPortHasBeenMapping = Boolean.FALSE;
    private Boolean taskSecurity = Boolean.TRUE;

    private List<String> classPathTaskFiles = new ArrayList<>();
    private List<TasksConfigProperties.TaskDeclare> taskDeclares = new ArrayList<>();

    private List<ScheduleTaskProxy> scheduleTaskProxies;
    private List<OnceTaskProxy> onceTaskProxies;

    public static final MethodHandles.Lookup lookup = FieldConvertUtil.methodLookup;
    private ThreadPoolExecutor threadPoolExecutor;

    private TaskAdapterService(AsyncTaskOperate asyncTaskOperate){
        this.restTemplate = new RestTemplate();
        this.asyncTaskOperate = asyncTaskOperate;
        init();
    };

    private void init(){
        String InvokeApi = (String)FieldOF.get_ofTp_fName_instance_.apply(
                TaskProxyHolder.class,
                "COMMON_INVOKE_API",
                null
        );
        FieldOF.set_ofTp_fName_val_.accept(
                TaskProxyHolder.class,
                "COMMON_INVOKE_API",
                String.format(InvokeApi, environment.getProperty(Const.PROP_SERVLET_CONTEXT, Strings.EMPTY))
        );
    }

    public static TaskAdapterService build(AsyncTaskOperate asyncTaskOperate){
        return new TaskAdapterService(asyncTaskOperate);
    }

    public TaskAdapterService threadPool(ThreadPoolExecutor threadPool){
        if(threadPool != null){
            this.threadPoolExecutor = threadPool;
        }
        return this;
    }

    public TaskAdapterService taskFilesResource(String... resourcesFiles){
        if(resourcesFiles != null && resourcesFiles.length > 0){
            Arrays.stream(resourcesFiles).filter(resourceFile -> Strings.isNotEmpty(resourceFile))
                    .forEach(resourceFile -> this.classPathTaskFiles.add(resourceFile));
        }
        return this;
    }

    public TaskAdapterService taskDeclares(TasksConfigProperties.TaskDeclare... taskDeclares){
        if(taskDeclares != null && taskDeclares.length > 0){
            Arrays.stream(taskDeclares).filter(taskDeclare -> taskDeclare != null)
                    .forEach(resourceFile -> this.taskDeclares.add(resourceFile));
        }
        return this;
    }

    public TaskAdapterService webPortHadBeenMapping(Boolean portHadBeenMapping){
        if(portHadBeenMapping != null){
            this.balanceServerPortHasBeenMapping = portHadBeenMapping;
        }
        return this;
    }

    public TaskAdapterService taskSecurity(Boolean security){
        if(security != null){
            this.taskSecurity = security;
        }
        return this;
    }

    private void parseTasksAndInit(){
        final String addressAndPort = String.join(":", hostAddress, port);
        final List<TasksConfigProperties.TaskDeclare> scheduleTasks = new ArrayList<>();
        final List<TasksConfigProperties.TaskDeclare> onceTasks = new ArrayList<>();

        // load tasks declared on resource files.
        if(!CommonCheck.nullList(classPathTaskFiles)){
            for (String filePosition : classPathTaskFiles) {
                String taskDeclaresMetadataContent = FileReadUtil.readClassPathFile(filePosition);
                Map<Boolean, List<TasksConfigProperties.TaskDeclare>> taskDeclaresGroup =
                        taskFilterAndPartition(taskDeclaresMetadataContent);
                scheduleTasks.addAll(taskDeclaresGroup.get(true));
                onceTasks.addAll(taskDeclaresGroup.get(false));
            }
        }

        // load tasks declared based Java API.
        if(!CommonCheck.nullList(taskDeclares)){
            String taskDeclaresMetadataContent = JSON.toJSONString(taskDeclares);
            Map<Boolean, List<TasksConfigProperties.TaskDeclare>> taskDeclaresGroup =
                    taskFilterAndPartition(taskDeclaresMetadataContent);
            taskDeclaresGroup.values().stream().flatMap(metadataGroup -> metadataGroup.stream())
                    .forEach(metadata -> rebuildParam(metadata));
            scheduleTasks.addAll(taskDeclaresGroup.get(true));
            onceTasks.addAll(taskDeclaresGroup.get(false));
        }

        // using config or using default threadPool.
        if(!CommonCheck.nullList(onceTasks) || !CommonCheck.nullList(scheduleTasks)){
            if(this.threadPoolExecutor != null){
                FieldOF.set_ofTp_fName_instance_val_.accept(
                        TaskProxyHolder.class,
                        "executorService",
                        null,
                        this.threadPoolExecutor
                );
            }
        }

        // init OnceTask.
        if(!CommonCheck.nullList(onceTasks)){
            onceTaskProxies = new ArrayList<>(onceTasks.size());
            for (TasksConfigProperties.TaskDeclare onceTask : onceTasks) {
                OnceTaskProxy onceTaskProxy = new OnceTaskProxy(
                        onceTask,
                        asyncTaskOperate,
                        restTemplate,
                        balanceServerPortHasBeenMapping,
                        taskSecurity,
                        addressAndPort
                );
                onceTaskProxies.add(onceTaskProxy);
            }
        }

        // init ScheduleTask.
        if(!CommonCheck.nullList(scheduleTasks)){
            scheduleTaskProxies = new ArrayList<>(scheduleTasks.size());
            for (TasksConfigProperties.TaskDeclare scheduleTask : scheduleTasks) {
                ScheduleTaskProxy scheduleTaskProxy = new ScheduleTaskProxy(
                        scheduleTask,
                        asyncTaskOperate,
                        restTemplate,
                        balanceServerPortHasBeenMapping,
                        taskSecurity,
                        addressAndPort
                );
                scheduleTaskProxies.add(scheduleTaskProxy);
            }
        }
    }

    // 对元数据task数组解析过滤并分组.
    private Map<Boolean, List<TasksConfigProperties.TaskDeclare>> taskFilterAndPartition(String taskMetadataResourceContent){
        List<TasksConfigProperties.TaskDeclare> tasks = JSON.parseArray(taskMetadataResourceContent, TasksConfigProperties.TaskDeclare.class);
        //获取适合的任务
        Map<Boolean, List<TasksConfigProperties.TaskDeclare>> suitableTaskPartition = tasks.stream().filter((taskUnit) -> {
            return ObjectOF.notNullAll_.test(taskUnit.strategy, taskUnit.className, taskUnit.methodName, taskUnit.params)
                    && taskUnit.params.stream().allMatch(param ->
                    ObjectOF.notNullAll_.test(param.paramClassName /*param.paramValue*/)
            );
        }).peek(taskUnit -> {
            if (Strings.isBlank(taskUnit.taskId)) {
                taskUnit.taskId = String.join("::", taskUnit.className, taskUnit.methodName);
            }
            if (taskUnit.strategy == TasksConfigProperties.TaskDeclare.TaskStrategy.SINGLE) {
                taskUnit.taskId = taskUnit.taskId + "@[single - *]";
            } else {
                taskUnit.taskId = taskUnit.taskId + "@[all - " + new SnowflakeIdUtil(1, 1).nextId() + "]";
            }
        }).collect(Collectors.partitioningBy((taskUnit) -> taskUnit.corn != null));

        return suitableTaskPartition;
    }

    //参数转换 同一转换自动配置来的字符串格式到 (JSON / JSONArray / Array / List / SpecialModel),
    //对于基础类型的数据结构和详细的参数类型转换由RunningInvoking调度时控制.
    private void rebuildParam(TasksConfigProperties.TaskDeclare metadata){
        for (TasksConfigProperties.TaskDeclare.Param param : metadata.params) {
            if(ObjectOF.notNull_.test(param.getParamValue()) &&
                    !ObjectOF.equals_.test(param.getParamClassName(),String.class.getName())){
                String paramVal = param.getParamValue().toString();
                String paramValClear = param.getParamValue().toString().replace("\n","").replace("\t","");
                if(JSON.isValidArray(paramValClear)){
                    param.setParamValue(JSON.parseArray(paramValClear));
                }else if(JSON.isValid(paramVal) && paramVal.startsWith("{") && paramVal.endsWith("}")){
                    param.setParamValue(JSON.parseObject(paramVal));
                }else if(ObjectOF.equals_.test(param.getParamClassName(),Object.class.getName())){
                    // Objects dealing ignore.
                }else {
                    //Base Type, dealing with RunningInvoking.
                }
            }
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        SureOF.notNull(restTemplate, "Con't instance the RestTemplate.");
        serviceId = valueResolver.resolveStringValue("${spring.application.name}");
        port = valueResolver.resolveStringValue("${server.port}");
        try {
            InetAddress address = InetAddress.getLocalHost();
            hostAddress = (address != null && address.getHostAddress() != null) ? address.getHostAddress() : "localhost";
        }catch (Throwable e) { }
        TaskProxyHolder.HTTP_PREFIX = String.join(Strings.EMPTY,"http://", hostAddress, ":" ,port);
        if(asyncTaskOperate instanceof AbstractRedisAsyncTask){
            ((AbstractRedisAsyncTask)asyncTaskOperate).clear();
        }
    }

    @Override
    public void afterSingletonsInstantiated() {
        Object retryLoadBalanceInterceptor = null;
        if(balanceServerPortHasBeenMapping) {
            // spring cloud 微服务环境下，采用服务负载的方式 需要引入此拦截器 负责choose broker
            // 为RestTemplate添加Ribbon拦截器，在请求链路中获取被容器映射过得真实的 node:port
            try {
                Class<?> retryLoadBalanceInterceptorClassMetadata =
                        Class.forName("org.springframework.cloud.client.loadbalancer.RetryLoadBalancerInterceptor");
                retryLoadBalanceInterceptor = beanFactory.getBean(retryLoadBalanceInterceptorClassMetadata);
                TaskProxyHolder.HTTP_PREFIX = String.join(Strings.EMPTY, "http://", serviceId);
            } catch (Throwable e) {
                log.info("\n  #### TaskAdapterService :: afterSingletonsInstantiated :: " +
                        "Can't find RetryLoadBalancerInterceptor, Task's loadBalance will be ignore, every request will be route to localhost...");
                retryLoadBalanceInterceptor = null;
            }
            //初始化 restTemplate, 初始化task
            try {
                Class<?> restTemplateMetadata = restTemplate.getClass();
                MethodHandle getInterceptorsHandle = lookup.findVirtual(restTemplateMetadata, "getInterceptors", MethodType.methodType(List.class));
                List templateBalanceInterceptors = (List)getInterceptorsHandle.bindTo(restTemplate).invoke();
                MethodHandle addHandle = lookup.findVirtual(List.class, "add", MethodType.methodType(boolean.class, Object.class));
                if(retryLoadBalanceInterceptor != null) {
                    addHandle.bindTo(templateBalanceInterceptors).invoke(retryLoadBalanceInterceptor);
                }
                addHandle.bindTo(templateBalanceInterceptors).invoke(new RibbonChooseResultInterceptor(asyncTaskOperate));
            }catch (Throwable e){
                throw new RuntimeException("TaskAdapterService :: afterSingletonsInstantiated :: occur an error, mayBe cause is restTemplate's interceptors is null.",e.getCause());
            }
        }
        //parse tasks and init.
        parseTasksAndInit();
    }

    @Override
    public void setEmbeddedValueResolver(StringValueResolver resolver) {
        this.valueResolver = resolver;
    }

    @Override
    public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
        this.beanFactory = beanFactory;
    }

    @Override
    public void setEnvironment(Environment environment) {
        this.environment = environment;
    }
}
